package org.apache.flink.connector.jdbc.catalog;

import org.apache.commons.compress.utils.Lists;
import org.apache.flink.connector.jdbc.dialect.JdbcDialectTypeMapper;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.UniqueConstraint;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.types.AbstractDataType;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TemporaryClassLoaderContext;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.PASSWORD;
import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.TABLE_NAME;
import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.URL;
import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.USERNAME;
import static org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory.IDENTIFIER;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;

//import org.apache.flink.connector.jdbc.databases.mysql.catalog.MySqlCatalog.fromJDBCType;

/**
 * @program: flink-connector-jdbc-3.1.1
 * @description: ${description}
 * @author: Copyright(c)2022-2032 carlos All Rights Reserved
 * @create by xiaolong.guo on 2023/10/20 15:03
 */
public class testPrimery {


    public static void main(String[] args) {
        String databaseName = "testdb";
        String dbUrl = "jdbc:mysql://hadoop105/testdb";
        String username = "root";
        String pwd = "root";
        String SchemaName = "testdb";
        String TableName = "books";
        ObjectPath tablePath = ObjectPath.fromString("testdb.books");

        try (Connection conn = DriverManager.getConnection(dbUrl, username, pwd)) {
            DatabaseMetaData metaData = conn.getMetaData();
            System.out.print( "连接信息:" + metaData + "\n");
            
            Optional<UniqueConstraint> primaryKey =
                    getPrimaryKey(
                            metaData,
                            databaseName,
                            SchemaName,
                            TableName);
            System.out.print( "主键:" + primaryKey  + "\n");

            //ps返回值：com.mysql.cj.jdbc.ClientPreparedStatement: SELECT * FROM books;
            PreparedStatement ps =
                    conn.prepareStatement(String.format("SELECT * FROM books;")); //, TableName

            System.out.print( "ps结果:" + ps  + "\n");

            // resultSetMetaData返回结果
            //com.mysql.cj.jdbc.result.ResultSetMetaData@5400db36 - Field level information: 
            //	com.mysql.cj.result.Field@76b74e9c[dbName=testdb,tableName=books,originalTableName=books,columnName=id,originalColumnName=id,mysqlType=3(FIELD_TYPE_INT),sqlType=4,flags= PRIMARY_KEY, charsetIndex=63, charsetName=ISO-8859-1]
            //	com.mysql.cj.result.Field@2d72f75e[dbName=testdb,tableName=books,originalTableName=books,columnName=title,originalColumnName=title,mysqlType=253(FIELD_TYPE_VARCHAR),sqlType=12,flags=, charsetIndex=45, charsetName=UTF-8]
            //	com.mysql.cj.result.Field@8ab78bc[dbName=testdb,tableName=books,originalTableName=books,columnName=authors,originalColumnName=authors,mysqlType=253(FIELD_TYPE_VARCHAR),sqlType=12,flags=, charsetIndex=45, charsetName=UTF-8]
            //	com.mysql.cj.result.Field@5aa0dbf4[dbName=testdb,tableName=books,originalTableName=books,columnName=year,originalColumnName=year,mysqlType=3(FIELD_TYPE_INT),sqlType=4,flags=, charsetIndex=63, charsetName=ISO-8859-1]

          
            ResultSetMetaData resultSetMetaData = ps.getMetaData();

            // [null, null, null, null]
            String[] columnNames = new String[resultSetMetaData.getColumnCount()];
            System.out.print( "columnNames的长度:" + columnNames.length  + "\n");
            
            // DataType[4]@3455
            DataType[] types = new DataType[resultSetMetaData.getColumnCount()];

            for (int i = 1; i <= resultSetMetaData.getColumnCount(); i++) {
                columnNames[i - 1] = resultSetMetaData.getColumnName(i);

                System.out.print(i + "--:--" + resultSetMetaData.getColumnName(i) + "\n");

//                types[i - 1] = fromJDBCType(tablePath, resultSetMetaData, i);
                
                if (resultSetMetaData.isNullable(i) == ResultSetMetaData.columnNoNulls) {
                    types[i - 1] = types[i - 1].notNull();
                }
            }

//            String[] columnNames = new String[]{"[id,title,authors,year]"};
//            DataType[] types = new DataType[]{}; //"int, varchar,varchar,int"

            Schema.Builder schemaBuilder = Schema.newBuilder().fromFields(columnNames, types);
            primaryKey.ifPresent(
                    pk -> schemaBuilder.primaryKeyNamed(pk.getName(), pk.getColumns()));
            Schema tableSchema = schemaBuilder.build();

            Map<String, String> props = new HashMap<>();
            props.put(CONNECTOR.key(), IDENTIFIER);
            props.put(URL.key(), dbUrl);
            props.put(USERNAME.key(), username);
            props.put(PASSWORD.key(), pwd);
            props.put(TABLE_NAME.key(), SchemaName);

            CatalogTable catalogTable = CatalogTable.of(tableSchema, null, Lists.newArrayList(), props);
            System.out.print("结果：" + catalogTable);
            // return CatalogTable.of(tableSchema, null, Lists.newArrayList(), props);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }

    protected static Optional<UniqueConstraint> getPrimaryKey(
            DatabaseMetaData metaData, String database, String schema, String table)
            throws SQLException {
        
        ResultSet rs = metaData.getPrimaryKeys(database, schema, table);
        
        
        Map<Integer, String> keySeqColumnName = new HashMap<>();
        
        String pkName = null;
        
        while (rs.next()) {
            String columnName = rs.getString("COLUMN_NAME");
            pkName = rs.getString("PK_NAME"); // all the PK_NAME should be the same
            int keySeq = rs.getInt("KEY_SEQ");
            
            Preconditions.checkState(!keySeqColumnName.containsKey(keySeq - 1),
                    "The field(s) of primary key must be from the same table.");
            
            keySeqColumnName.put(keySeq - 1, columnName); // KEY_SEQ is 1-based index
        }
        
        List<String> pkFields = Arrays.asList(new String[keySeqColumnName.size()]); // initialize size
        
        keySeqColumnName.forEach(pkFields::set);
        
        if (!pkFields.isEmpty()) {
            // PK_NAME maybe null according to the javadoc, generate an unique name in that case
            pkName = pkName == null ? "pk_" + String.join("_", pkFields) : pkName;
            return Optional.of(UniqueConstraint.primaryKey(pkName, pkFields));
            
            // Optional[CONSTRAINT `PRIMARY` PRIMARY KEY (`id`) NOT ENFORCED]
        }
        
        return Optional.empty();
    }

    protected String getSchemaName(ObjectPath tablePath) {
        throw new UnsupportedOperationException();
    }

    protected String getSchemaTableName(ObjectPath tablePath) {
        throw new UnsupportedOperationException();
    }

    public testPrimery fromFields(String[] fieldNames, AbstractDataType<?>[] fieldDataTypes) {
        Preconditions.checkNotNull(fieldNames, "Field names must not be null.");
        Preconditions.checkNotNull(fieldDataTypes, "Field data types must not be null.");
        Preconditions.checkArgument(
                fieldNames.length == fieldDataTypes.length,
                "Field names and field data types must have the same length.");
        
        return this;
    }

    protected DataType fromJDBCType(ObjectPath tablePath, ResultSetMetaData metadata, int colIndex)
            throws SQLException {
        throw new UnsupportedOperationException();
    }
    
}
